package net;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.data.Stat;

public class DistributedLock implements Watcher {
	private ZooKeeper zk;
	private String root = "/locks";// 根
	private String waitNode;// 等待节点
	private String myZnode;// 当前节点
	private CountDownLatch latch;// 计数器
	private CountDownLatch connectLatch;
	private int sessionTimeout = 30000;// 超时时间

	private Logger logger = Logger.getLogger(DistributedLock.class);

	public DistributedLock(String config) {
		try {
			connectLatch = new CountDownLatch(1);
			zk = new ZooKeeper(config, sessionTimeout, this);
			if (States.CONNECTING == zk.getState()) {
				connectLatch.await();
				Stat stat = zk.exists(root, false);
				if (stat == null) {
					zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
							CreateMode.PERSISTENT);
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
			logger.error(e);
		} catch (KeeperException e) {
			e.printStackTrace();
			logger.error(e);
		} catch (InterruptedException e) {
			e.printStackTrace();
			logger.error(e);
		}
	}

	/**
	 * zookeeper节点的监视器
	 */
	@Override
	public void process(WatchedEvent event) {
		if (event.getType() == EventType.None
				&& event.getState() == KeeperState.SyncConnected) {
			connectLatch.countDown();
		} else if (event.getType() == Event.EventType.NodeDeleted) {
			if (this.latch != null) {
				this.latch.countDown();
			}
		}
	}

	public void lock() throws InterruptedException, KeeperException {
		if (this.tryLock()) {
			System.out.println("Thread " + Thread.currentThread().getId() + " "
					+ myZnode + " get lock true");
			return;
		} else {
			waitForLock(waitNode);// 等待锁
			System.out.println("Thread " + Thread.currentThread().getId() + " "
					+ myZnode + " get waitlock[" + waitNode + "] true");
		}
	}

	private boolean tryLock() throws KeeperException, InterruptedException {
		myZnode = zk.create(root + "/lock_", new byte[0],
				ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

		List<String> list = zk.getChildren(root, false);
		Collections.sort(list);
		if (myZnode.equals(root + "/" + list.get(0))) {
			return true;
		} else {
			String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);
			waitNode = list.get(Collections.binarySearch(list, subMyZnode) - 1);
		}
		return false;
	}

	private boolean waitForLock(String lower) throws InterruptedException,
			KeeperException {
		Stat stat = zk.exists(root + "/" + lower, true);
		// 判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听
		if (stat != null) {
			System.err.println(myZnode + "等待节点:" + lower);
			this.latch = new CountDownLatch(1);
			this.latch.await(sessionTimeout, TimeUnit.MILLISECONDS);
			this.latch = null;
		}
		return true;
	}

	public void unlock() throws InterruptedException, KeeperException {
		zk.delete(myZnode, -1);
		myZnode = null;
		zk.close();
	}

}